#include "server.h"
#include <atmi.h>

#define _CONN_TIMEOUT           10          // 连接读写超时
#define _SESS_TIMEOUT           60          // 前台接入超时
#define _TRANS_TIMEOUT          60          // 事务超时

#define _RECVBUF_DEF_SIZE       KB(1)
#define _RETURN_CODE_SIZE       4

enum { _COMMIT_, _ROLLBACK_ };

struct worker_ctl
{/*{{{*/
    unsigned int thrd_id;
    int worker_idx;
    EP ep;
    EP_SLOT listener_slot;
    char *request;
    char *respond;

    unsigned int seq;
    unsigned int tux_connected;
    SVCNODE *svcnode;
};/*}}}*/
static struct worker_ctl *a_worker_ctl;

struct _svctrack
{/*{{{*/
    struct _svc
    {
        char name[NAME_SIZE+1];
        char begin_ts_s[TIMESTAMP_SIZE+1];
        char end_ts_s[TIMESTAMP_SIZE+1];
    } *svcs;
    int svc_alloc_num;
    int svc_num;
    time_t begin_ts;
    char begin_ts_s[TIMESTAMP_SIZE+1];
    time_t end_ts;
    char end_ts_s[TIMESTAMP_SIZE+1];
};/*}}}*/

static void _svctrack_init(struct _svctrack *svctrack)
{/*{{{*/
    svctrack->svcs=(struct _svc *)NULL;
    svctrack->svc_alloc_num=svctrack->svc_num=0;
    svctrack->begin_ts=curr_timestamp(svctrack->begin_ts_s);
    svctrack->end_ts=0;
    CLR_BUF(svctrack->end_ts_s);
}/*}}}*/

static int _svctrack_begin(struct _svctrack *svctrack, char *svc)
{/*{{{*/
#define _ALLOC_NUM      5
    struct _svc *svcp;

    if(svctrack->svc_num == svctrack->svc_alloc_num)
    {
        EXPAND_ARRAY_DOERR(svctrack->svcs, struct _svc,
                _ALLOC_NUM, svctrack->svc_alloc_num,
                errlog("%s: expand svctrack.svcs fail", __func__);
                return(-1));
    }
    svcp=&svctrack->svcs[svctrack->svc_num];
    svctrack->svc_num++;
    strcpy(svcp->name, svc);
    curr_timestamp(svcp->begin_ts_s);
    CLR_BUF(svcp->end_ts_s);
    return(0);
}/*}}}*/

static void _svctrack_end(struct _svctrack *svctrack)
{/*{{{*/
    curr_timestamp(svctrack->svcs[svctrack->svc_num-1].end_ts_s);
}/*}}}*/

static void _svctrack_dump(struct worker_ctl *worker_ctl,
        struct _svctrack *svctrack, unsigned int abort)
{/*{{{*/
    if(abort)
    {
        char buf[BUFFER_SIZE];
        int offset=0;
        int i;
        struct _svc *svcp;

        svctrack->end_ts=curr_timestamp(svctrack->end_ts_s);
        offset+=sprintf(buf+offset,
                "%s: worker(%d) sess(%d) use %ld secs, begin(%s)->",
                __func__,
                worker_ctl->worker_idx, worker_ctl->seq,
                svctrack->end_ts-svctrack->begin_ts,
                svctrack->begin_ts_s+DATE_SIZE);
        for(i=0; i<svctrack->svc_num; i++)
        {
            svcp=&svctrack->svcs[i];
            offset+=sprintf(buf+offset, "%s(%s-%s)->",
                    svcp->name, svcp->begin_ts_s+DATE_SIZE,
                    svcp->end_ts_s+DATE_SIZE);
        }
        sprintf(buf+offset, "end(%s)", svctrack->end_ts_s+DATE_SIZE);
        errlog(buf);
    }
    FREE(svctrack->svcs);
}/*}}}*/

static void _sess_disconn(struct worker_ctl *worker_ctl,
        SOCKET new_sock, unsigned int client_call)
{/*{{{*/
    struct _worker_info *worker=
        &stat_info._worker_.worker[worker_ctl->worker_idx];
    sock_close(new_sock);
    pthread_mutex_lock(&stat_info._worker_.lock);
    --worker->conn_num;
    ++worker->client_count;
    worker->client_call+=client_call;
    pthread_mutex_unlock(&stat_info._worker_.lock);
}/*}}}*/

static int _sess_read(struct worker_ctl *worker_ctl, SOCKET new_sock)
{/*{{{*/
    char *request=worker_ctl->request;
    int request_size;
    int res;

    res=sock_read(new_sock, (unsigned char *)request, &request_size,
            _CONN_TIMEOUT, WRAP);
    if(res)
    {
        int offset=0;
        char buf[BUFFER_SIZE];

        offset+=sprintf(buf, "%s: worker(%d:%d) peer ",
                __func__,
                worker_ctl->worker_idx, worker_ctl->seq);
        if(res == RET_ERR_ABORT)
            offset+=sprintf(buf+offset, "abort");
        else if(res == RET_ERR_TIMEOUT)
            offset+=sprintf(buf+offset, "timeout");
        else
            offset+=sprintf(buf+offset, "read fail, res:%d",
                    res);
        errlog(buf);
        return(-1);
    }
    request[request_size]=ZERO;
    return(0);
}/*}}}*/

static int _sess_write(struct worker_ctl *worker_ctl, SOCKET new_sock)
{/*{{{*/
    char *respond=worker_ctl->respond;
    int respond_size;
    int res;

    respond_size=strlen(respond);
    res=sock_write(new_sock, (unsigned char *)respond, &respond_size,
            _CONN_TIMEOUT, WRAP);
    if(res)
    {
        errlog("%s: worker(%d:%d) respond write fail, res:%d",
                __func__,
                worker_ctl->worker_idx, worker_ctl->seq, res);
        return(-1);
    }
    return(0);
}/*}}}*/

static int _tux_conn(struct worker_ctl *worker_ctl)
{/*{{{*/
    if(!worker_ctl->tux_connected)
    {
        TPINIT *tpinitbuf;
        tpinitbuf=(TPINIT *)tpalloc("TPINIT", NULL, TPINITNEED(16));
        tpinitbuf->flags=TPMULTICONTEXTS;
        if(tpinit(tpinitbuf) == -1)
        {
            errlog("%s: worker(%d:%d) tpinit fail: %d %s, BAD",
                    __func__, worker_ctl->worker_idx, worker_ctl->seq,
                    tperrno, tpstrerror(tperrno));
            tpfree((char *)tpinitbuf);
            return(-1);
        }
        tpfree((char *)tpinitbuf);
        worker_ctl->tux_connected=1;
#ifdef _MY_DEBUG
        errlog("%s: worker(%d) connect",
                __func__, worker_ctl->worker_idx);
#endif
    }
    return(0);
}/*}}}*/

static void _tux_disconn(struct worker_ctl *worker_ctl)
{/*{{{*/
    if(tpterm() == -1)
    {
        errlog("%s: worker(%d:%d) tpterm fail: %d %s, BAD",
                __func__,
                worker_ctl->worker_idx, worker_ctl->seq,
                tperrno, tpstrerror(tperrno));
    }
    worker_ctl->tux_connected=0;
#ifdef _MY_DEBUG
        errlog("%s: worker(%d) disconnect",
                __func__, worker_ctl->worker_idx);
#endif
}/*}}}*/

static int _trans_begin(struct worker_ctl *worker_ctl)
{/*{{{*/
    // 连接Tuxedo
    if(_tux_conn(worker_ctl))
    {
        return(-1);
    }

    // 开始事务
#ifdef _MY_DEBUG 
    errlog("%s: worker(%d) before begin",
            __func__, worker_ctl->worker_idx);
#endif
    if(tpbegin(_TRANS_TIMEOUT, 0) == -1)
    {
        errlog("%s: worker(%d:%d) tpbegin fail: %d %s, BAD",
                __func__, worker_ctl->worker_idx,
                worker_ctl->seq,
                tperrno, tpstrerror(tperrno));
        _tux_disconn(worker_ctl);
        return(-1);
    }
#ifdef _MY_DEBUG
    errlog("%s: worker(%d) begin", __func__, worker_ctl->worker_idx);
#endif
    return(0);
}/*}}}*/

static int _trans_end(struct worker_ctl *worker_ctl, unsigned int type)
{/*{{{*/
    int res;

    switch(type)
    {
        case _COMMIT_: res=tpcommit(0); break;
        case _ROLLBACK_: res=tpabort(0); break;
    }
    if(res == -1)
    {
        _tux_disconn(worker_ctl);
        return(-1);
    }
    return(0);
}/*}}}*/

static int _svcenv(struct worker_ctl *worker_ctl,
        struct _svctrack *svctrack,
        char *request, char *respond)
{/*{{{*/
    int worker_idx=worker_ctl->worker_idx;
    char *body;
    char *sendbuf;
    long sendlen;
    char *recvbuf;
    long recvlen;
    char *svc;

    body=NULL;
    if(!(svc=para_get_item(request, &body)))
    {
        errlog("%s: worker(%d) svc null", __func__, worker_idx);
        para_put_item(respond, SOAR_ERR_SVC_INVALID);
        return(0);
    }

    _svctrack_begin(svctrack, svc);

    sendbuf=recvbuf=NULL;
    sendlen=strlen(request);
    recvlen=_RECVBUF_DEF_SIZE;
    if(!(sendbuf=(char *)tpalloc("STRING", NULL, sendlen+1)) ||
            !(recvbuf=(char *)tpalloc("STRING", NULL, recvlen)))
    {
        errlog("%s: worker(%d:%d) tpalloc fail",
                __func__, worker_idx, worker_ctl->seq);
        if(sendbuf) tpfree(sendbuf);
        if(recvbuf) tpfree(recvbuf);
        para_put_item(respond, SOAR_ERR_RESOURCE);
        return(-1);
    }
    strcpy(sendbuf, body);
    if(tpcall(svc, sendbuf, 0L, &recvbuf, &recvlen, 0L) == -1)
    {
        //int tpcallerr=tperrno;
        errlog("%s: worker(%d:%d) tpcall(%s) fail: %d %s",
                __func__, worker_idx, worker_ctl->seq,
                svc, tperrno, tpstrerror(tperrno));
        tpfree(sendbuf); tpfree(recvbuf);
        para_put_item(respond, SOAR_ERR_SVC_NOT_FOUND);
        return(-1);
    }

    if(recvlen < OPTION_SIZE+PARA_ITEM_SIZE_CHAR+1+
            _RETURN_CODE_SIZE+1)
    {
        errlog("%s: worker(%d) recvlen(%ld) < %d+1+%d",
                __func__, worker_idx, recvlen,
                PARA_ITEM_SIZE_CHAR, _RETURN_CODE_SIZE);
        para_put_item(respond, SOAR_ERR_SVC_NOT_FOUND);
        return(-1);
    }
    if(recvbuf[PARA_ITEM_SIZE_CHAR+1] == '9')
    {
        errlog("%s: worker(%d) server error code: %.4s",
                __func__, worker_idx,
                recvbuf+PARA_ITEM_SIZE_CHAR+1);
    }
    strcpy(respond, recvbuf);
    tpfree(sendbuf);
    tpfree(recvbuf);
#ifdef _MY_DEBUG
    errlog("%s: worker(%d) call finished, respond_size(%d)",
            __func__, worker_idx, strlen(respond));
#endif

    _svctrack_end(svctrack);

    return(0);
}/*}}}*/

static int _svcnode(struct worker_ctl *worker_ctl,
        char *request, char *respond)
{/*{{{*/
    char *curr;
    char *snname;
    int size;
    int sn_idx;
    struct _cnfg_svcnode *cnfg_sn;
    int res;

    curr=NULL;
    if(!(snname=para_get_item(request, &curr)) ||
            (size=strlen(snname)) == 0 ||
            size > NAME_SIZE)
    {
        errlog("%s %s: svcnode(%s) invalid",
                __func__, snname?snname:"null");
        para_put_item(respond, SOAR_ERR_PARA);
        return(-1);
    }
    for(sn_idx=0; sn_idx<cnfg_info._svcnode_.num; ++sn_idx)
        if(strcmp(cnfg_info._svcnode_.svcnode[sn_idx].name, snname)
                == 0)
            break;
    if(sn_idx == cnfg_info._svcnode_.num)
    {
        errlog("%s: svcnode(%s) not found",
                __func__, snname);
        para_put_item(respond, SOAR_ERR_SVC_NOT_FOUND);
        return(-1);
    }
    cnfg_sn=&cnfg_info._svcnode_.svcnode[sn_idx];
#ifdef _MY_DEBUG
    errlog("%s: svcnode(%d:%s)",
            __func__, sn_idx, cnfg_sn->name);
#endif
    if(!worker_ctl->svcnode[sn_idx])
    {
        if((res=sn_create(cnfg_sn->addr, cnfg_sn->mode,
                &worker_ctl->svcnode[sn_idx])))
        {
            errlog("%s: svcnode(%s) create fail",
                    __func__, snname);
            para_put_item(respond, SOAR_ERR_RESOURCE);
            return(-1);
        }
    }
    res=sn_call(worker_ctl->svcnode[sn_idx], NULL, curr, respond);
    if(res)
    {
        errlog("%s: worker(%d): sn(%s) sn_call fail, res:%d",
                __func__, worker_ctl->worker_idx, cnfg_sn->name, res);
        return(-1);
    }

    return(0);
}/*}}}*/

static int _cache(struct worker_ctl *worker_ctl,
        char *request, char *respond)
{/*{{{*/
    para_put_item(respond, SOAR_ERR_OK);
    return(0);
}/*}}}*/

static int _loop(struct worker_ctl *worker_ctl, char *newconn)
{/*{{{*/
    int worker_idx=worker_ctl->worker_idx;
    char *request=worker_ctl->request;
    char *respond=worker_ctl->respond;
    char para_desc[100];
    PARA para;
    SOCKET new_sock;
    time_t access_ts;
    time_t now;
    unsigned int end=IS_FALSE;
    unsigned int abort=IS_FALSE;
    unsigned int trans_type=_ROLLBACK_;
    struct _svctrack svctrack;
    unsigned int client_call=0;
    char *cmd;
    char *body;
    char *value;
    int res;

    sprintf(para_desc, "new_sock(number:1-%d) access_ts(number:1-%d)",
            SHORT_NUMBER_SIZE, LONG_FILE_NM_SIZE);
    para_load(newconn, para_desc, &para);
    para_ref_by_name(para, "new_sock", &value);
    new_sock=atoi(value);
    para_ref_by_name(para, "access_ts", &value);
    access_ts=atol(value);
    para_free(para);

    // 判断接入是否超时
    now=curr_timestamp(NULL);
    res=now-access_ts;
    if(res > _SESS_TIMEOUT)
    {
        errlog("%s: worker(%d) now("LONG_NUMBER_FMT")"
                "-access("LONG_NUMBER_FMT")=(%d) timeout",
                __func__, worker_idx, now,
                access_ts, res);
        _sess_disconn(worker_ctl, new_sock, client_call);
        return(-1);
    }

    if(++worker_ctl->seq == 10000)
        worker_ctl->seq=0;

    _svctrack_init(&svctrack);

    if(_trans_begin(worker_ctl))
    {
        CLR_BUF(respond);
        para_put_item(respond, SOAR_ERR_RESOURCE);
        _sess_write(worker_ctl, new_sock);
        goto GO_OUT;
    }

    while(!end)
    {/*{{{*/
        // 读取
        res=_sess_read(worker_ctl, new_sock);
        if(res)
        {
            abort=IS_TRUE;
            break;
        }
#ifdef _MY_DEBUG
        errlog("%s: _sess_read", __func__);
#endif

        // 解析服务
        CLR_BUF(respond);
        body=NULL;
        cmd=para_get_item(request, &body);
        if(!cmd)
        {
            errlog("%s: worker(%d) cmd null",
                    __func__, worker_idx);
            goto RESPOND;
        }
        if(strcasecmp(cmd, "RELEASE") == 0 ||
                strcasecmp(cmd, "RELEASE") == 0)
        {
            end=IS_TRUE;
            trans_type=_COMMIT_;
            para_put_item(respond, SOAR_ERR_OK);
#ifdef _MY_DEBUG
            errlog("%s: %s", __func__, cmd);
#endif
        }
        else if(strcasecmp(cmd, "ABORT") == 0)
        {
            end=IS_TRUE;
            trans_type=_ROLLBACK_;
            para_put_item(respond, SOAR_ERR_OK);
#ifdef _MY_DEBUG
            errlog("%s: %s", __func__, cmd);
#endif
        }
        else if(strcasecmp(cmd, "CALL") == 0)
        {
            _svcenv(worker_ctl, &svctrack, body, respond);
        }
        else if(strcasecmp(cmd, "SVCNODE") == 0)
        {
            _svcnode(worker_ctl, body, respond);
        }
        else if(strcasecmp(cmd, "CACHE") == 0)
        {
            _cache(worker_ctl, body, respond);
        }
        else
        {
            errlog("%s: worker(%d) not CALL", __func__, worker_idx);
            para_put_item(respond, SOAR_ERR_CMD_INVALID);
        }

RESPOND:
        // 返回
        if(_sess_write(worker_ctl, new_sock))
        {
            abort=IS_TRUE;
            break;
        }
        ++client_call;
    }/*}}}*/

    if(_trans_end(worker_ctl, trans_type))
    {
        CLR_BUF(respond);
        para_put_item(respond, SOAR_ERR_RESOURCE);
        _sess_write(worker_ctl, new_sock);
        goto GO_OUT;
    }

    _svctrack_dump(worker_ctl, &svctrack, abort);

GO_OUT:
    _sess_disconn(worker_ctl, new_sock, client_call);
    return(0);
}/*}}}*/

static int _read(struct worker_ctl *worker_ctl,
        struct ep_active_slot *active_slot)
{/*{{{*/
    char *request=active_slot->_u._read.msg;
    char *body;
    char *svc;
    int res;

    request[active_slot->_u._read.msg_size]=ZERO;
    body=request;
    if(!(svc=para_get_item(request, &body)) ||
            (res=strlen(svc)) <= 0 || res > NAME_SIZE)
    {
        errlog("%s: svc(%s) invalid, impossible",
                __func__, (svc)?svc:SPACES);
        return(-1);
    }

    if(strcmp(svc, "NEWCONN") == 0)
    {
        _loop(worker_ctl, body);
    }
    else
    {
        errlog("%s: svc(%s) not found, impossible",
                __func__, svc);
        return(-1);
    }

    return(0);
}/*}}}*/

static void _worker(struct worker_ctl *worker_ctl)
{/*{{{*/
//#define UPDATE_STAT_INFO_INTERVAL           10
//#define COUNT_NUM                           1000000
#define CHECK_OPERATE_INFO_INTERVAL         1
    int worker_idx=worker_ctl->worker_idx;
    struct ep_active_slot *active_list;
    unsigned int quit=IS_FALSE;
    time_t now;
//    time_t last_stat=0;
    time_t last_operate=0;
    int res;

    while(!quit)
    {
        if((res=ep_poll(worker_ctl->ep, &active_list)) != 0)
        {
            errlog("%s: worker(%d) revoke listener, res:%d, "
                    "impossible",
                    __func__, worker_idx, res);
            continue;
        }
        while(active_list)
        {
            switch(active_list->type)
            {
                case EP_SLOT_READ:
                    _read(worker_ctl, active_list);
                    break;
                default:
                    errlog("%s: worker(%d) poll type(%s) invalid, "
                            "impossible",
                            __func__, worker_idx, active_list->type);
                    break;
            }
            EP_ACTIVE_LIST_NEXT(active_list);
        }

        now=curr_timestamp(NULL);

#if 0
        // update stat_info.client calls
        if(now - last_stat > UPDATE_STAT_INFO_INTERVAL)
        {/*{{{*/
            struct _worker_info *worker=
                &stat_info._worker_.worker[worker_idx];

            pthread_mutex_lock(&stat_info._worker_.lock); //lock
            if(worker_ctl->client_count > 0)
            {
                worker->client_count+=worker_ctl->client_count;
                if(worker->client_count >= COUNT_NUM)
                {
                    worker->client_loop+=
                        worker->client_count/COUNT_NUM;
                    worker->client_count%=COUNT_NUM;
                }
                worker_ctl->client_count=0;
            }
            pthread_mutex_unlock(&stat_info._worker_.lock); //unlock

            last_stat=now;
        }/*}}}*/
#endif

        // check operate_info 
        if(now - last_operate > CHECK_OPERATE_INFO_INTERVAL)
        {/*{{{*/
            pthread_mutex_lock(&operate_info._worker_.lock);
            if((quit=operate_info._worker_.quit))
            {
                operate_info._worker_.state[worker_idx]=THRD_SHUTDOWN;
            }
            pthread_mutex_unlock(&operate_info._worker_.lock);

            last_operate=now;
        }/*}}}*/
    }
    DEEP_SLEEP;
}/*}}}*/

int worker_start(SOCKET **p_sock_pair)
{/*{{{*/
    SOCKET *sock_pair;
    struct worker_ctl *worker_ctl;
    int worker_idx;
    int res;

    // stat_info
    pthread_mutex_init(&stat_info._worker_.lock, NULL);
    ALLOC_MULTI_DOERR(stat_info._worker_.worker, struct _worker_info,
            cnfg_info.threads,
            fprintf(stderr,
                "agent: alloc stat_info._worker_.worker fail\n");
            return(-1));
    for(worker_idx=0; worker_idx<cnfg_info.threads; ++worker_idx)
    {
        struct _worker_info *worker=&stat_info._worker_.worker[worker_idx];
        worker->conn_num=0;
        worker->client_count=0;
        worker->client_call=0;
    }

    // operate_info
    pthread_mutex_init(&stat_info._worker_.lock, NULL);
    operate_info._worker_.quit=IS_FALSE;
    ALLOC_MULTI_DOERR(operate_info._worker_.state, enum thrd_state,
            cnfg_info.threads,
            fprintf(stderr,
                "agent: alloc operate_info._worker_.state fail\n");
            return(-1));
    for(worker_idx=0; worker_idx<cnfg_info.threads; ++worker_idx)
    {
        operate_info._worker_.state[worker_idx]=THRD_RUNNING;
    }

    // start
    ALLOC_MULTI_DOERR(sock_pair, SOCKET, cnfg_info.threads*2,
            fprintf(stderr, "agent: alloc sock_pair fail\n");
            return(-1));
    ALLOC_MULTI_DOERR(a_worker_ctl, struct worker_ctl, cnfg_info.threads,
            fprintf(stderr, "agent: alloc a_worker_ctl fail\n");
            return(-1));
    for(worker_idx=0; worker_idx<cnfg_info.threads; ++worker_idx)
    {
        int i;

        // open listener-worker channel
        if(sock_local_pair(&sock_pair[worker_idx*2]))
        {
            fprintf(stderr,
                    "agent: open listener-worker socket pair fail\n");
            return(-1);
        }
        worker_ctl=&a_worker_ctl[worker_idx];
        if((res=ep_create(&worker_ctl->ep)))
        {
            fprintf(stderr,
                    "agent: worker(%d)(ep_slot_register) listener fail, "
                    "errcode:%d\n",
                    worker_idx, res);
            return(-1);
        }
        if((res=ep_slot_register(worker_ctl->ep,
                        sock_pair[worker_idx*2+1], IS_FALSE,
                        &worker_ctl->listener_slot)))
        {
            fprintf(stderr,
                    "agent: worker(%d)(ep_slot_register) listener fail, "
                    "errcode:%d\n",
                    worker_idx, res);
            return(-1);
        }
        ep_slot_set_wrap(worker_ctl->listener_slot);

        worker_ctl->worker_idx=worker_idx;
        ALLOC_MULTI_DOERR(worker_ctl->request, char, SOCK_MSG_SIZE,
                fprintf(stderr, "agent: alloc worker(%d) request fail\n",
                    worker_idx);
                return(-1));
        ALLOC_MULTI_DOERR(worker_ctl->respond, char, SOCK_MSG_SIZE,
                fprintf(stderr, "agent: alloc worker(%d) respond fail\n",
                    worker_idx);
                return(-1));
        worker_ctl->seq=0;
        worker_ctl->tux_connected=0;
        ALLOC_MULTI_DOERR(worker_ctl->svcnode, SVCNODE,
            cnfg_info._svcnode_.num,
            fprintf(stderr, "agent: alloc worker(%d) svcnode fail\n",
                worker_idx);
            return(-1));
        for(i=0; i<cnfg_info._svcnode_.num; ++i)
            worker_ctl->svcnode[i]=NULL;

        if(thrd_create((THRD_FUNC_T)_worker, (void *)worker_ctl,
                    THRD_DETACH, &worker_ctl->thrd_id))
        {
            fprintf(stderr, "agent: worker(%d) thrd create fail\n",
                    worker_idx);
            return(-1);
        }
    }

    *p_sock_pair=sock_pair;
    return(0);
}/*}}}*/
